package bigoo.peer;

import static java.util.logging.Logger.GLOBAL_LOGGER_NAME;

import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

import bigoo.common.PeerInfo;
import bigoo.common.communication.AbstractCommunication;
import bigoo.common.communication.Message;
import bigoo.common.communication.MessageReceiver;
import bigoo.ds.DiscoveryService;
import bigoo.ds.message.GetPeersAnswerMessage;
import bigoo.ds.message.ImAliveAnswerMessage;
import bigoo.peer.message.AcceptRideAnswerMessage;
import bigoo.peer.message.AcceptRideMessage;
import bigoo.peer.message.ConfirmRideMessage;
import bigoo.peer.message.GetPeersMessage;
import bigoo.peer.message.ImAliveMessage;
import bigoo.peer.message.PublishRideMessage;
import bigoo.peer.message.RejectedRideMessage;
import bigoo.peer.message.RideRequestMessage;
import bigoo.peer.ride.NoTransactionException;
import bigoo.peer.ride.Request;
import bigoo.peer.ride.RequestManager;
import bigoo.peer.ride.Ride;
import bigoo.peer.ride.RideManager;
import bigoo.peer.ride.Request.RequestState;
import bigoo.util.Config;

public class BigooPeer implements MessageReceiver, PeerFailureInterested {

    private Logger logger = Logger.getLogger(GLOBAL_LOGGER_NAME);
    private Properties props = Config.getProperties();
    private AbstractCommunication communication;
    private ScheduledExecutorService scheduler = null;
    private Map<String, PeerInfo> neighbors = new HashMap<String, PeerInfo>(); // peerIDxpeerInfo
    private String myPeerID;
    private String ipAddress;
    private PeerInfo ds;
    private RideManager rideMng;
    private RequestManager requestMng;
    private int lease = 30000;
    private int threshold = 1;
    private ScheduledFuture<?> epidemicFuture = null;

    public BigooPeer(AbstractCommunication communication, String peerId, String ipAddress) {
	this.scheduler = Executors.newSingleThreadScheduledExecutor();

	this.communication = communication;
	this.communication.addReceiver(this);
	this.communication.addPeerInterested(this);

	this.myPeerID = peerId;
	this.ipAddress = ipAddress;

	String dsAddr = props.getProperty(Config.PEER_DS_ADDRESS);
	int dsPort = Integer.valueOf(props.getProperty(Config.PEER_DS_PORT));
	this.ds = new PeerInfo(DiscoveryService.DS_ID, dsAddr, dsPort);
	this.rideMng = new RideManager();
	this.requestMng = new RequestManager();
	this.communication.addPeerInterested(this.requestMng);

	this.communication.start();
    }

    public void start() {
	this.scheduler.scheduleAtFixedRate(new RideRequester(), 5000, 5000, TimeUnit.MILLISECONDS);
	this.sendImAliveMessage();
    }

    public void peerIsUp(String peerId, String ipAddress, int port) {
	logger.info("New neighbor peer \'" + peerId + "\' is up.");
	PeerInfo newPeer = new PeerInfo(peerId, ipAddress, port);
	neighbors.put(peerId, newPeer);
    }

    @Override
    public void peerFailed(String peerID) {
	if (peerID.equals(DiscoveryService.DS_ID)) {
	    logger.info("Discovery service is not up. Trying to contact DS again in " + lease
		    + "milliseconds. Using last lease and threshold values.");
	    scheduleTasks();
	} else {
	    logger.info("Neighbor peer \'" + peerID + "\' has failed.");
	    neighbors.remove(peerID);
	    this.requestMng.peerFailed(peerID);
	}
    }

    @Override
    public void processMessage(Message msg) {
	String msgID = msg.getMessageIdentifier();

	if (msg.getSenderID().equals(DiscoveryService.DS_ID)) {
	    if (msgID.equals(ImAliveAnswerMessage.MSG_IDENTIFIER)) {
		processImAliveAnswer((ImAliveAnswerMessage) msg);
	    } else if (msgID.equals(GetPeersAnswerMessage.MSG_IDENTIFIER)) {
		processGetPeersAnswer((GetPeersAnswerMessage) msg);
	    }
	} else {
	    if (msgID.equals(PublishRideMessage.MSG_IDENTIFIER)) {
		processPublishRide((PublishRideMessage) msg);
	    } else if (msgID.equals(RideRequestMessage.MSG_IDENTIFIER)) {
		processRideRequest((RideRequestMessage) msg);
	    } else if (msgID.equals(AcceptRideMessage.MSG_IDENTIFIER)) {
		processAcceptRide((AcceptRideMessage) msg);
	    } else if (msgID.equals(AcceptRideAnswerMessage.MSG_IDENTIFIER)) {
		processAcceptRideAnswer((AcceptRideAnswerMessage) msg);
	    } else if (msgID.equals(ConfirmRideMessage.MSG_IDENTIFIER)) {
		processConfirmRide((ConfirmRideMessage) msg);
	    } else if (msgID.equals(RejectedRideMessage.MSG_IDENTIFIER)) {
		processRejectedRide((RejectedRideMessage) msg);
	    }
	}
    }

    private void processImAliveAnswer(ImAliveAnswerMessage m) {
	logger.info("IM_ALIVE_ANSWER message received from " + m.getSenderID());
	if (this.neighbors.size() < m.getThreshold()) {
	    this.sendGetPeersMessage();
	} else {
	    this.communication.closeConnection(ds);
	}
	lease = m.getLease();
	threshold = m.getThreshold();
	this.scheduleTasks();
    }

    private void processGetPeersAnswer(GetPeersAnswerMessage m) {
	logger.info("GET_PEERS_ANSWER message received from " + m.getSenderID());
	for (PeerInfo peer : m.getPeers()) {
	    this.neighbors.put(peer.getPeerID(), peer);
	}
	this.communication.closeConnection(ds);
    }

    private void processPublishRide(PublishRideMessage m) {
	String peerID = m.getSenderID();
	logger.info("PUBLISH_RIDE message received from " + peerID);
	// let's merge rides!
	int merged = this.rideMng.mergeRides(m.getRides());
	logger.fine(merged + " new rides merged!");
	// if i've never talked to this guy before, register it!
	if (!neighbors.containsKey(peerID)) {
	    peerIsUp(peerID, m.getSenderIpAddress(), m.getSenderPort());
	}
    }

    private void processRideRequest(RideRequestMessage m) {
	// TODO notify ui!
	System.out.println("New ride request received from " + m.getSenderID() + " to ride " + m.getRideID());
	Ride ride = this.rideMng.getRide(m.getRideID());
	if (ride != null) {
	    PeerInfo peer = new PeerInfo(m.getSenderID(), m.getSenderAddress(), m.getSenderPort());
	    String requestID = this.requestMng.registerOtherRequest(ride, peer);
	    logger.info("Ride request received from " + peer.getPeerID() + " for ride " + ride.getRideID());
	    logger.fine("The requestID is " + requestID);
	} else {
	    logger.warning("Request received for unknown ride: " + m.getRideID());
	}
    }

    private void processAcceptRide(AcceptRideMessage m) {
	try {
	    logger.info("AcceptRide received from " + m.getSenderID() + " for request " + m.getRequestID());
	    Request req = this.requestMng.getRequest(m.getRequestID());
	    if (req != null && req.getRequestState() == RequestState.OPEN) {
		this.requestMng.openTransaction(req);
		String requestID = m.getRequestID();
		this.requestMng.changeRequestToNextState(req);
		AcceptRideAnswerMessage message = new AcceptRideAnswerMessage(this.myPeerID, requestID);
		this.sendMessage(message, req.getPeer());
		this.requestMng.changeRequestToNextState(req);
	    }
	} catch (NoTransactionException e) {
	    logger.warning(e.getLocalizedMessage());
	}
    }

    private void processAcceptRideAnswer(AcceptRideAnswerMessage m) {
	try {
	    logger.info("AcceptAnswer received from " + m.getSenderID() + " for request " + m.getRequestID());
	    Request req = this.requestMng.getRequest(m.getRequestID());
	    if (req != null && req.getRequestState() == RequestState.WAITING) {
		String requestID = m.getRequestID();
		ConfirmRideMessage message = new ConfirmRideMessage(this.myPeerID, requestID);
		if (this.sendMessage(message, req.getPeer()))
		    this.rideMng.markOnePlaceAsUsed(req.getRideID());
		this.requestMng.changeRequestToNextState(req);
		this.requestMng.closeTransaction(req);
	    }
	} catch (NoTransactionException e) {
	    logger.warning(e.getLocalizedMessage());
	}
    }

    private void processConfirmRide(ConfirmRideMessage m) {
	try {
	    logger.info("ConfirmRide received from " + m.getSenderID() + " for request " + m.getRequestID());
	    Request req = this.requestMng.getRequest(m.getRequestID());
	    if (req != null && req.getRequestState() == RequestState.WAITING) {
		this.requestMng.changeRequestToNextState(req);
		System.out.println("Ride confirmed: " + req);
	    } else {
		logger.finest(req.getRequestState().toString());
	    }
	} catch (NoTransactionException e) {
	    logger.warning(e.getLocalizedMessage());
	}
    }

    private void processRejectedRide(RejectedRideMessage m) {
	logger.info("Request " + m.getRequestID() + " rejected!!");
	Request req = this.requestMng.getRequest(m.getRequestID());
	if (req != null) {
	    if (req.getRequestState() != RequestState.CONFIRMED)
		this.requestMng.requestRejected(req);
	    else
		logger.warning("Reject receive for a already confirmed request!!!!! " + req.getRequestID());
	}
    }

    private void scheduleTasks() {
	logger.fine("Next ImAlive will be fired in " + lease + " milliseconds");
	logger.fine("Next PublishRide will be fired in " + lease / threshold + " milliseconds");
	if (epidemicFuture != null) {
	    epidemicFuture.cancel(false);
	}
	this.scheduler.schedule(new ImAliveTrigger(), lease, TimeUnit.MILLISECONDS);
	epidemicFuture = this.scheduler.scheduleAtFixedRate(new EpidemicAgent(), 0, (lease / threshold), TimeUnit.MILLISECONDS);
    }

    private void sendImAliveMessage() {
	logger.info("Sending \"I'm alive\" message to discovery service...");
	Message imAlive = new ImAliveMessage(myPeerID, ipAddress, props.getProperty(Config.PEER_PORT));
	this.sendMessageToDiscoverySevice(imAlive);
    }

    private void sendGetPeersMessage() {
	logger.info("Requesting peers to discovery service..");
	Message getPeers = new GetPeersMessage(myPeerID);
	this.sendMessageToDiscoverySevice(getPeers);
    }

    private void sendMessageToDiscoverySevice(Message message) {
	this.sendMessage(message, ds);
    }

    private boolean sendMessage(Message message, PeerInfo peer) {
	try {
	    this.communication.sendMessage(message, peer);
	    return true;
	} catch (IOException e) {
	    logger.warning("Error on sending message to " + peer.getPeerID() + ": " + e.getLocalizedMessage());
	    return false;
	}
    }

    // methods called from Controller/UI

    public Collection<Request> getMyRequests() {
	return this.requestMng.getMyRequests();
    }

    public Collection<Request> getOthersRequests() {
	return this.requestMng.getOthersRequests();
    }

    public Collection<Request> getConfirmedRequests() {
	return this.requestMng.getConfirmedRequests();
    }

    public void publishRide(String destination, int places, String date) {
	int port = Integer.parseInt(props.getProperty(Config.PEER_PORT));
	this.rideMng.registerRide(destination, this.myPeerID, this.ipAddress, port, places, date);
    }

    public Collection<Ride> getAvailableRiders() {
	return this.rideMng.getAvailableRiders();
    }

    public void requestRide(String rideID) throws IllegalArgumentException {
	Ride ride = this.rideMng.getRide(rideID);
	if (ride != null) {
	    this.requestMng.registerMyRequest(ride, this.myPeerID);
	} else {
	    logger.warning("There's no ride with id " + rideID);
	    throw new IllegalArgumentException("There's no ride with id " + rideID);
	}
    }

    public void acceptRequest(String requestID) throws IllegalArgumentException {
	try {
	    Request req = this.requestMng.getRequest(requestID);
	    if (req != null && req.getRequestState() == RequestState.OPEN) {
		logger.info("Accept Ride for request " + requestID);
		this.requestMng.openTransaction(req);
		this.requestMng.changeRequestToNextState(req);
		AcceptRideMessage m = new AcceptRideMessage(this.myPeerID, requestID);
		this.sendMessage(m, req.getPeer());
		this.requestMng.changeRequestToNextState(req);
	    } else {
		logger.warning("There's no request with id " + requestID);
	    }
	} catch (NoTransactionException e) {
	    logger.warning(e.getLocalizedMessage());
	}
    }

    public void rejectRequest(String requestID) throws IllegalArgumentException {
	Request req = this.requestMng.getRequest(requestID);
	if (req != null && req.getRequestState() == RequestState.OPEN) {
	    logger.info("Reject Ride for request " + requestID);
	    this.requestMng.requestRejected(req);
	    RejectedRideMessage m = new RejectedRideMessage(this.myPeerID, requestID);
	    this.sendMessage(m, req.getPeer());
	} else {
	    logger.warning("There's no request with id " + requestID);
	}
    }

    private class ImAliveTrigger implements Runnable {
	@Override
	public void run() {
	    sendImAliveMessage();
	}
    }

    private class EpidemicAgent implements Runnable {
	@Override
	public void run() {
	    if (neighbors.size() > 0) {
		PeerInfo peer = (PeerInfo) neighbors.values().toArray()[(int) Math.round(Math.random() * (neighbors.size() - 1))];
		Collection<Ride> rides = rideMng.getSomeRides(peer.getPeerID());
		PublishRideMessage message = new PublishRideMessage(myPeerID, ipAddress, Integer.valueOf(props.getProperty(Config.PEER_PORT)), rides);
		logger.info("Sending rides to " + peer.getPeerID());
		sendMessage(message, peer);
	    }
	}
    }

    private class RideRequester implements Runnable {
	@Override
	public void run() {
	    int port = Integer.parseInt(props.getProperty(Config.PEER_PORT));
	    Collection<Request> requests = requestMng.getDisconnectedRequests();
	    for (Request request : requests) {
		RideRequestMessage m = new RideRequestMessage(myPeerID, ipAddress, port, request.getRideID());
		logger.info("Trying to establish contact with provider for request " + request.getRequestID());
		requestMng.openRequest(request);
		sendMessage(m, request.getPeer());
	    }
	}
    }

}
